package com.xian.java.batch;


import com.mongodb.hadoop.io.BSONWritable;
import com.mongodb.hadoop.mapred.MongoInputFormat;
import com.xian.java.bean.traffic.Ws_traffic_operate_pay;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.hadoop.mapred.HadoopInputFormat;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.hadoop.mapred.JobConf;
import org.bson.BSONObject;

/**
 * flink获取mongo数据
 */
public class GetMongoData2 {
    public static void main(String[] args) throws Exception {

        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        //String mongoInputUri = parameterTool.get("mongoInputUri");
        //String mongoInputUri = "mongodb://pubuser:KAov0XvKeFPVrF+4+rz-6P49B9U6Rz@172.18.4.25:20017/activity.activityAssistanceRelationPo?authSource=admin";  //指定验证库正确方式
        //String mongoInputUri = "mongodb://pubuser:KAov0XvKeFPVrF+4+rz-6P49B9U6Rz@172.18.4.25:20017/activity.activityAssistanceRelationPo?authSource=activity";  //指定验证库正确方式
        String mongoInputUri = "mongodb://pubuser:KAov0XvKeFPVrF+4+rz-6P49B9U6Rz@172.18.4.25:20017/traffic.ws_traffic_operate_pay?authSource=admin";  //指定验证库正确方式
        System.out.println("传入mongoInputUri="+mongoInputUri);

        // 获取执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 定义输入格式
        HadoopInputFormat<BSONWritable, BSONWritable> hdIf = new HadoopInputFormat<BSONWritable, BSONWritable>(
                new MongoInputFormat(), BSONWritable.class, BSONWritable.class,	new JobConf());

        // 指定mongo 连接url
        hdIf.getJobConf().set("mongo.input.uri", mongoInputUri);

        DataSource<Tuple2<BSONWritable, BSONWritable>> dataSource = env.createInput(hdIf);

        dataSource.print();
        long count = dataSource.count();
        System.out.println("总共数据量为:"+count);

        MapOperator<Tuple2<BSONWritable, BSONWritable>, Object> map = dataSource.map(new RichMapFunction<Tuple2<BSONWritable, BSONWritable>, Object>() {
            @Override
            public Object map(Tuple2<BSONWritable, BSONWritable> value) throws Exception {
                System.out.println(value);
                System.out.println("             ");
                System.out.println(value.f0);
                System.out.println();
                System.out.println(value.f1);
                System.out.println();
                System.out.println(value.f1.getDoc());

                BSONObject bsonObject = value.f1.getDoc();
                Object aClass = bsonObject.get("_class");
                System.out.println(aClass+"\n");

                Object data = bsonObject.get("data");
                System.out.println(data+"\n");


                BSONObject dataBJ = ((BSONObject) data);

                Object orderId = dataBJ.get("orderId");
                System.out.println(orderId+"\n");

                Object orderTime = dataBJ.get("orderTime");
                System.out.println(orderTime+"\n");

                Object o = dataBJ.get("startPoint.lat");
                System.out.println(o);


                return null;
            }
        });


        map.count();








        System.out.println("程序运行完毕！");
    }
}
